
----------------------------------------------------------------
--- Copyright：  迷你玩-海马
--- Description：（只提供给 data_cache服务 ）
--- dc读写share_dict的封装，其他业务应该不会使用到这个文件
----------------------------------------------------------------


local assert     = assert
local require    = require    
local bit        = bit
local coroutine  = coroutine
local debug      = debug
local io         = io
local pairs      = pairs
local ipairs     = ipairs
local math       = math
local os         = os
local print      = print
local pcall      = pcall
local xpcall     = xpcall
local rawget     = rawget
local rawset     = rawset
local select     = select
local string     = string
local table      = table
local tonumber   = tonumber
local tostring	 = tostring
local error      = error
local type       = type
local unpack     = unpack
local setmetatable = setmetatable
local getmetatable = getmetatable
local ngx = ngx

local table_insert = table.insert
local table_remove = table.remove
local ngx_sleep = ngx.sleep
local ngx_now   = ngx.now

require "resty.core.shdict"

local ngx_shared  = ngx.shared;
local string_sub  = string.sub
local string_gsub = string.gsub
local string_format = string.format

local log       = require("modules.commhm.log");
local ns_time   = require("modules.commhm.time");
local uu        = require("modules.commhm.utils");


local __TJ_INTERVAL__      = 180


local NGX_NULL  = "not_found"                  --ngx.null
local NGX_AUTH  = "authentication required."


-- shm MGR类
local _M = { _VERSION = '0.12.1' }
local mt = { __index = _M }


function _M.new(self, pre_ )
	log.debug( log.fast() and "call shmMgr:new" );
	local ins_ = {
		config_cache  = {};
		pre           = pre_;

		select_mod    = 10;      ---10分表 (目前只支持10分表)

		last_time_report_tj        = uu.now() + __TJ_INTERVAL__;

		shm_delay_clear = {};

		get_set_tj   = {};
		-- get_call
		-- get_cache
		-- get_cache_stale
		-- get_shm
		-- set_shm
		-- delay_shm

		mread_tj = {
			call  = 0,
			get   = 0,
			gets  = 0,
			set   = 0,
		};

		_lock_id   = (ngx.worker.id() + 1) * 1000;    --lock_id

		_cc        = 0;   --计数
		_last_time = uu.now();
	};
    return setmetatable(ins_, mt);
end


_M.init = function(self)
end


_M.final = function(self)
	self:do_report_tj();
end


-------------------------------------------------------
-- 出错处理
_M.error_exit = function(self, txt, trace_info_ )
	if  ngx.ctx.err_call_back then
		--log.error( "has set error_exit. " .. (txt or 'nil') );
		ngx.ctx.err_call_back(txt, trace_info_);
	else
		log.error( "not set error_exit. " .. (txt or 'nil'), trace_info_ );
	end
end


_M.close = function( self, shutdown )
	--log.debug( log.fast() and "shm keep or close all ok." );
end


-------------------------------------------------------------------------------
_M.set_get_set_tj = function(self, hash_, name_)
	if     ngx.ctx.m_params.act then
		hash_ = hash_ .. "_" .. ngx.ctx.m_params.act
	elseif ngx.ctx.m_params.cmd then
		hash_ = hash_ .. "_" .. ngx.ctx.m_params.cmd
	end

	if  not self.get_set_tj[hash_] then
		self.get_set_tj[hash_] = {
			get_call  = 0,
			get_cache = 0,
			get_cache_stale  = 0,
			get_shm = 0,
			set_shm = 0,
			delay_shm = 0,
		};
	end

	if  self.get_set_tj[hash_][ name_ ] then
		self.get_set_tj[hash_][ name_ ] = self.get_set_tj[hash_][ name_ ] + 1
	else
		log.error( "call set_get_set_tj, unknown name_=" .. name_ )
	end

end


_M.do_report_tj = function(self)
	local now_ = uu.now()
	if  now_ > self.last_time_report_tj then
		self.last_time_report_tj = now_ + __TJ_INTERVAL__;

		--开始打印统计
		for hash_, v1 in pairs(self.get_set_tj) do
			if  v1.get_call + v1.set_shm > 0 then
				log.day_list_short( "dc_shm", "cache_hit_shm", "get_set", self.pre,
						"get", v1.get_call, v1.get_cache,  v1.get_cache_stale, v1.get_shm,
						"set", v1.set_shm, v1.delay_shm, hash_ )
			end
		end
		self.get_set_tj = {}

		if  self.mread_tj.call + self.mread_tj.get + self.mread_tj.set > 0 then
			log.day_list_short( "dc_shm", "cache_hit_shm", "mread", self.pre, self.mread_tj.call,
					self.mread_tj.get, self.mread_tj.gets, self.mread_tj.set )
			self.mread_tj = { call=0, get=0, gets=0, set=0 }
		end
	end
end


---进行shm key_hash锁检查
_M.check_shm_resty_lock = function(self, select_, key_hash_)
	local dict_ = ngx_shared[self.pre .. select_]
	if  dict_ then
		--2秒中内循环试图锁定一个 key_hash_lk ( exists则等待 )
		local timeout_ = 2000   ---毫秒
		local step_    = 1      ---步长毫秒
		local cc_      = 0

		---计算锁id
		self._lock_id = self._lock_id + 1
		if  self._lock_id >= 100000 then self._lock_id = 1000 end

		while timeout_ > 0 do
			cc_ = cc_ + 1
			local ok, err = dict_:add( key_hash_ .. "_lk" , self._lock_id, 3)      ---锁住主键kh  12345_6_lk
			if  ok then
				if  cc_ > 10 then
					--循环等待其他锁的次数太多
					log.day_list_short( "shm", "lock_rw", "lock_ok0", ngx.worker.id(), uu.get_act_cmd(),
							key_hash_, cc_, 2000-timeout_, step_, self._lock_id );
				end
				return self._lock_id  --成功
			else
				if  err == 'exists' then
					timeout_ = timeout_ - step_
					if  timeout_ <= 0 then
						log.day_list_short("shm", "lock_rw", "error_time_out0",
								ngx.worker.id(), uu.get_act_cmd(), self.pre, select_,
								key_hash_, cc_, timeout_, step_, debug.traceback() )
						---完全超时 read请求并没有发出去，也没有开始connect，不断开其他链接
						return 0, "error_time_out_lock"
					end

					if  cc_ <= 10 then
						step_ = 50 + ( cc_ * 2)   ---等待时间逐步放大   500+2+4+6+8+10+12+14+16+18+20=500 + 110
					else
						step_ = (20-cc_) * 10      ---等待时间逐步缩小  100 90 80 70 60 50 40 30
						if  step_ < 10 then step_ = 10 end
					end

					--log.day_list_short( "shm", "lock_rw", "step0", ngx.worker.id(), uu.get_act_cmd(),
						--key_hash_, timeout_, step_, debug.traceback() );
					ngx_sleep( step_ * 0.001 )

				else
					log.day_list_short( "shm", "lock_rw", "error_step0",
							ngx.worker.id(), uu.get_act_cmd(), self.pre, select_,
							key_hash_, cc_, timeout_, step_, err, debug.traceback() );
					return 0, err   ---其他错误
				end
			end
		end
	else
		if  self.select_mod ~= 10 then
			---提示可能是百分表
			log.error( "error_select: plz check select_mod=" .. self.select_mod )
		end

		log.day_list_short("shm", "lock_rw", "error_no_dict", ngx.worker.id(), uu.get_act_cmd(),
				self.pre, select_, key_hash_, debug.traceback() );
		return 0, "error_no_ngx_shared"   ---找不到 ngx_shared
	end

	return 0, "unknown_err"   ---未知错误 (应该不能运行到这里，健壮性保证)
end


---远程调用解锁，只有解锁后，其他协程才能继续获得读写锁
---read_only_ 不解锁，等待final统一解锁
_M.unlock_resty_lock = function( self, select_, key_hash_, read_only_ )
	if  read_only_ then
		return "ok"  ---不解锁
	end

	local dict_ = ngx_shared[ self.pre .. select_ ]
	if  dict_ then
		local ok, err_ = dict_:delete( key_hash_ .. "_lk")
		if  ok then
			--删除成功
			--log.day_list_short( "shm", "lock_rw", "unlock_ok", ngx.worker.id(), uu.get_act_cmd(), key_hash_ );
			return "ok"
		else
			log.day_list_short( "shm", "lock_rw", "error_delete_unlock1", ngx.worker.id(),
					uu.get_act_cmd(), key_hash_, err_, debug.traceback() );
			return "not_lock"
		end
	else
		log.error( "dict_not_exist=" .. self.pre .. select_ )
		log.day_list_short( "shm", "lock_rw", "error_dict_not_exist",  self.pre .. select_ );
		return "no_dict"
	end

end


-------------------------------------------------------------------------------
--- 读请求 ---
--- p_stale_ 接受过期数据
_M.API_shm_read = function( self, p_select_, key, hash, p_stale_, err_call_back, read_only_ )
	log.debug( log.fast() and "call API_shm_read, select=" .. p_select_
			.. ", key=" .. key .. ", hash=" .. hash );
	ngx.ctx.err_call_back = err_call_back;

	local select_ = p_select_ % self.select_mod

	self:set_get_set_tj( hash, "get_call" )
	local key_hash_ = key .. "_h" .. hash

	--乐观锁r
	if  read_only_ then
		--只读接口
	else
		local lock_id_, err_msg_ = self:check_shm_resty_lock(select_, key_hash_)
		if  lock_id_ == 0 then
			---获得锁失败 一般是超时 read请求并没有发出去，也没有开始connect，不断开其他链接
			self:error_exit( err_msg_ );
			ngx.exit(ngx.ERROR);
			return
		end
		ngx.ctx.__lock_id__ = lock_id_   ---记录lock_id
	end

	local ret2, stale_ = self:get_shared_cache_shm( select_, key_hash_ )
	if  ret2 then
		--log.debug( log.fast() and "ret2 cache hit." )
		if  stale_ then
			self:set_get_set_tj( hash, "get_cache_stale" )
			if  not ('1' == p_stale_) then
				return NGX_NULL;
			end
		else
			self:set_get_set_tj( hash, "get_cache" )
		end
		return ret2   --cache命中
	end

	return NGX_NULL;
end



-------------------------------------------------------
_M.API_shm_save = function( self, p_select_, key, hash, value, et_, err_call_back, read_only_ )
	log.debug( log.fast() and "call API_shm_save, select=" .. p_select_
			.. ", key=" .. key .. ", hash=" .. hash );

	local select_ = p_select_ % self.select_mod
	ngx.ctx.err_call_back = err_call_back;

	local key_hash_ = key .. "_h" .. hash
	--优先写入share cache
	local ret_ = self:set_shared_cache_shm( select_, key_hash_, value, et_ )
	if  ret_ then
		self:unlock_resty_lock( select_, key_hash_, read_only_ )
		self:set_get_set_tj( hash, "delay_shm" )
		return "OK"  --缓写落地数据
	end

	return "OK"
end



-- multi read
-- p_stale_ 接受过期数据
_M.API_shm_mread = function( self, p_select_, all_kv_list, p_stale_, err_call_back )
	log.debug( log.fast() and "call API_shm_mread, select=" .. p_select_ );

	local select_ = p_select_ % self.select_mod
	ngx.ctx.err_call_back = err_call_back;

	local fret = {};
	local fret_stdmap = {};   ---帮助排序结构体

	--先查cache
	for i=1, #all_kv_list do
		self.mread_tj.call = self.mread_tj.call + 1

		local key_hash_ = all_kv_list[i].key .. "_h" .. all_kv_list[i].hash

		fret_stdmap[ key_hash_ ] = all_kv_list[i]     ---fret_stdmap和all_kv_list映射起来 [ table --指针--> table ]

		local ret2, stale_ = self:get_shared_cache_shm( select_, key_hash_ )
		if  ret2 then
			--命中
			if  stale_ then
				self.mread_tj.gets = self.mread_tj.gets + 1
				if  '1' == p_stale_ then
					fret_stdmap[ key_hash_ ].v = ret2
				end
			else
				self.mread_tj.get  = self.mread_tj.get + 1
				fret_stdmap[ key_hash_ ].v = ret2   ---记录结果   ---设置fret_stdmap等于设置all_kv_list
			end
		end
	end

	--- 最后合并结果
	for i=1, #all_kv_list do
		if  all_kv_list[i] then
			if  not all_kv_list[i].v then
				all_kv_list[i].v = NGX_NULL
			end
			fret[ i ] = all_kv_list[i]
		end
	end

	return fret;
end



-- 删除一个key下的一个hash
_M.API_shm_del = function( self, p_select_, key, hash, err_call_back )
	log.debug( log.fast() and "call API_shm_del, select=" .. p_select_
			.. ", key=" .. key .. ", hash=" .. hash );

	local select_ = p_select_ % self.select_mod
	ngx.ctx.err_call_back = err_call_back;

	--self, select_, key_, hash_ )
	self:clean_shared_cache_shm( select_, key, hash )
	return 0;
end


-- 主动加锁
_M.API_shm_try_get_resty_lock = function(self, p_select_, key_hash_ )
	local select_ = p_select_ % self.select_mod
	return self:check_shm_resty_lock(select_, key_hash_ )
end


-- 主动解锁
_M.API_shm_unlock_resty_lock = function( self, p_select_, key_hash_, err_call_back )
	ngx.ctx.err_call_back = err_call_back;
	local select_ = p_select_ % self.select_mod
	return self:unlock_resty_lock( select_, key_hash_ )
end



-----------------------------ngx_shared--------------------------------

--设置全局数据缓存kv 十分表
_M.set_shared_cache_shm = function( self, select_, key_, str_, expire_ )
	if  select_ and key_ and str_  then

		local dict_name_ = self.pre ..  select_
		local share_ = ngx_shared[ dict_name_ ];
		if  share_ then

			if  expire_ and type(expire_) == 'string' then
				--log.debug( log.fast() and "expire_=" .. expire_ .. " k=" .. key_ .. " v=" .. str_ )
				expire_ = tonumber(expire_) or 0
			end

			if  expire_ and expire_ > 0 and expire_ < 86400*30 then
				--normal
			else
				expire_ = 86400*30     --30*60
			end

			local ok, err, forcible = share_:set( key_, str_, expire_ );
			if  ok then
				if  forcible then
					log.day_list_short("dc_shm", "cache_hit_shm", "set_forcible", dict_name_, key_, err )
				end
				return true
			else
				if  err then
					--share_:delete( key_ )
					--set出错，立即删除缓存
					local ok, err, forcible = share_:set( key_, nil );
					if  err or forcible then
						log.day_list_short("dc_shm", "cache_hit_shm", "error_delete", dict_name_, key_, err, forcible )
					end
				end
				log.day_list_short("dc_shm", "cache_hit_shm", "error_set", dict_name_, key_, err )
			end
		else
			log.day_list_short("dc_shm", "cache_hit_shm", "error_set_no_dict", dict_name_, key_)
		end
	end

	return false   --写入cache失败
end



--获得全局数据缓存kv 十分表
_M.get_shared_cache_shm = function( self, select_, key_ )
	if  select_ and key_ then

		local dict_name_ = self.pre ..  select_
		local share_ = ngx_shared[ dict_name_ ];
		if  share_ then

			local str_, f_, stale_ = share_:get_stale( key_);
			--uu.report_share_free_space( share_, dict_name_, key_, "dc_shm" )

			if  str_ then
				return str_, stale_  	--正常返回
			else
				--有报错的情况 ret = nil, error
				if  f_ then
					log.day_list_short("dc_shm", "cache_hit_shm", "error_get_nil", dict_name_, key_, f_ )
				end
			end
		else
			log.day_list_short("dc_shm", "cache_hit_shm", "error_get_no_dict", dict_name_, key_)
		end
	end

	return nil, nil
end


--清理掉一个cache数据 十分表
_M.clean_shared_cache_shm = function(self, select_, key_, hash_ )
	if  select_ and key_ and hash_ then
		local dict_name_ = self.pre ..  select_

		local share_ = ngx_shared[ dict_name_ ];
		if  share_ then
			local ok, err, forcible = share_:set( key_ .. "_h" .. hash_ , nil );
			if  err or forcible then
				log.day_list_short("dc_shm", "cache_hit_shm", "error_delete", dict_name_, key_, err, forcible )
			end
		end
	end
end






return _M;
